package com.bungarus.busi;

import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.model.MessageBuilder;
import com.bungarus.model.MessageType;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.apache.log4j.Logger;

/**
 * Created by tang on 2018/11/16.
 */
public class NodeAsClientMessageHandler extends ChannelInboundHandlerAdapter {
    Logger logger = Logger.getLogger(NodeAsClientMessageHandler.class);
    private final Callback<MessageProtobuf.Message, Void> persistOfflineCallback;

    public NodeAsClientMessageHandler(Callback<MessageProtobuf.Message, Void> callback) {
        this.persistOfflineCallback = callback;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if(msg instanceof MessageProtobuf.Message){
            MessageProtobuf.Message message = (MessageProtobuf.Message)msg;
            switch (message.getHeader().getType()) {
                case MessageType.ACK:
                    Discharger.discharge(message.getHeader().getId());
                    break;
                case MessageType.HEARTBEAT_PONG:
                    break;
                case MessageType.OFFLINE:
                    //Access node could send back message tagged with OFFLINE to busi node.
                    //Persist the offline message
                    if(null == persistOfflineCallback) {
                        throw new IllegalStateException("The callback to persist offline message is required.");
                    }
                    persistOfflineCallback.call(message);
                    ctx.writeAndFlush(MessageBuilder.buildAckMessage(message.getHeader().getId(), message.getHeader().getId2()));
                    logger.debug("============== offline message: " + message.toString());
                    break;
                default:
                    break;
            }
        }
        else {
            super.channelRead(ctx, msg);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        if(cause instanceof IllegalStateException) {
            //send exception message which don't need be ack
            ctx.writeAndFlush(MessageBuilder.buildPersistCallbackRequiredMessage());
        }
    }
}
